package org.apache.solr.ocean.core;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.FilterIndexReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.search.ExtendedFieldCache;
import org.apache.lucene.store.instantiated.InstantiatedIndex;
import org.apache.lucene.store.instantiated.InstantiatedIndexReader;
import org.apache.lucene.store.instantiated.InstantiatedIndexWriter;
import org.apache.solr.ocean.Documents;
import org.apache.solr.ocean.cache.Cache;
import org.apache.solr.ocean.cache.CacheKey;
import org.apache.solr.ocean.cache.DocSetCacheKey;
import org.apache.solr.ocean.util.ConcurrentTreeMap;
import org.apache.solr.ocean.util.Constants;
import org.apache.solr.ocean.util.Timeout;
import org.apache.solr.schema.IndexSchema;
import org.apache.solr.search.DocSet;

// TODO: release old MemoryIndexSnapshots from map
public class WriteableMemoryIndex extends Index {
	private WriteableIndexWriter indexWriter;
	private InstantiatedIndexReader indexReader;
	private InstantiatedIndex instantiatedIndex;
	private ConcurrentTreeMap<Long, MemoryIndexSnapshot> snapshotMap = new ConcurrentTreeMap<Long, MemoryIndexSnapshot>();
	private Cache cache;

	public WriteableMemoryIndex(Long id, IndexSchema indexSchema, TransactionSystem.Category category) throws IOException {
		super(id, indexSchema, category);
		instantiatedIndex = new InstantiatedIndex();
		indexWriter = new WriteableIndexWriter(instantiatedIndex);
		indexReader = new InstantiatedIndexReader(instantiatedIndex);
		cache = initCache();
	}

	// called by Category.runTransactionsNotInIndex
	void addDocuments(Documents documents, Analyzer analyzer) throws IOException {
		for (Document document : documents) {
			indexWriter.addDocument(document, analyzer);
		}
	}

	// called by Category.runTransactionsNotInIndex
	MemoryIndexSnapshot setSnapshot(Long snapshotId, List<Deletes> deletesList, IndexSchema schema) throws Exception {
		int maxDoc = indexReader.maxDoc();
		HashSet<Integer> deletedSet = new HashSet<Integer>();
		if (deletesList != null) {
			for (Deletes deletes : deletesList) {
				applyDeletes(false, deletes, null, indexReader, schema);
			}
		}
		MemoryIndexSnapshot memoryIndexSnapshot = new MemoryIndexSnapshot(snapshotId, maxDoc, deletedSet, schema);
		snapshotMap.put(snapshotId, memoryIndexSnapshot);
		return memoryIndexSnapshot;
	}

	public MemoryIndexSnapshot getIndexSnapshot(Long snapshotId) {
		return snapshotMap.get(snapshotId);
	}

	public MemoryIndexSnapshot getLatestIndexSnapshot() {
		return snapshotMap.lastValue();
	}

	public class MemoryIndexSnapshot extends IndexSnapshot {
		private final int maxDoc;
		private HashSet<Integer> deletedDocs;
		private SnapshotIndexReader snapshotIndexReader;

		public MemoryIndexSnapshot(Long snapshotId, int maxDoc, HashSet<Integer> deletedDocs, IndexSchema schema) {
			super(snapshotId, schema);
			this.maxDoc = maxDoc;
			this.deletedDocs = deletedDocs;
			snapshotIndexReader = new SnapshotIndexReader(maxDoc - deletedDocs.size());
		}
		
		public boolean containsKey(CacheKey cacheKey) {
			return cache.containsKey(cacheKey);
		}
		
		public int maxDoc() {
		  return maxDoc;
		}
		
		public Set<CacheKey> getCacheKeys() {
			return cache.getCacheKeys();
		}

		public IndexReader getIndexReader() {
			return snapshotIndexReader;
		}

		public class SnapshotIndexReader extends FilterIndexReader {
			private int numDocs;
			
			public SnapshotIndexReader(int numDocs) {
				super(indexReader);
				this.numDocs = numDocs;
			}

			public boolean isDeleted(int doc) {
				if (doc >= maxDoc) return true;
				return deletedDocs.contains(doc);
			}

			public boolean hasDeletions() {
				return true;
			}

			public int numDocs() {
				return numDocs;
			}

			public int maxDoc() {
				return maxDoc;
			}
		}

		public DocSet getDocSet(DocSetCacheKey cacheKey, Timeout timeout) throws Exception {
			return cacheKey.createObject(this, timeout);
		}
	}

	Long getLatestSnapshotID() {
		return snapshotMap.lastKey();
	}

	private HashSet<Integer> getLatestSnapshotDeletedDocSet() {
		MemoryIndexSnapshot memoryIndexSnapshot = snapshotMap.lastValue();
		if (memoryIndexSnapshot == null || memoryIndexSnapshot.deletedDocs == null)
			return null;
		HashSet<Integer> deletedDocSet = memoryIndexSnapshot.deletedDocs;
		return deletedDocSet;
	}

	public int getDocumentCount() {
		return indexReader.maxDoc();
	}

	public DeletesResult commitDeletes(Deletes deletes, Transaction transaction) throws InterruptedException, IOException {
		return commitChanges(null, deletes, null, transaction);
	}

	public DeletesResult commitChanges(Documents documents, Deletes deletes, Analyzer analyzer, Transaction transaction) throws InterruptedException, IOException {
		try {
			if (isClosed()) {
				throw new IOException("index is closed");
			}
			if (isReadOnly() && documents != null && documents.size() > 0) {
				throw new IOException("index not accepting new documents");
			}
			DeletesResult deletesResult = new DeletesResult(getId());
			HashSet<Integer> deletedSet = null;
			if (deletes != null && deletes.hasDeletes()) {
				HashSet<Integer> previousDeletedSet = getLatestSnapshotDeletedDocSet();
				if (previousDeletedSet != null) {
					deletedSet = (HashSet<Integer>) previousDeletedSet.clone();
				} else {
					deletedSet = new HashSet<Integer>();
				}
				IndexSnapshot indexSnapshot = this.getLatestIndexSnapshot();
				deletesResult = applyDeletes(false, deletes, deletedSet, indexReader, transaction.getIndexSchema());
			} else if (deletes == null || !deletes.hasDeletes()) { // if no deletes
				// just use same
				deletedSet = getLatestSnapshotDeletedDocSet();
			}
			if (documents != null) {
				for (Document document : documents) {
					indexWriter.addDocument(document, analyzer);
				}
			}
			transaction.ready(this);
			if (transaction.go()) {
				indexWriter.commit();
				int maxDoc = indexReader.maxDoc();
				Long snapshotId = transaction.getId();
				MemoryIndexSnapshot memoryIndexSnapshot = new MemoryIndexSnapshot(snapshotId, maxDoc, deletedSet, transaction.getIndexSchema());
				snapshotMap.put(transaction.getId(), memoryIndexSnapshot);
				return deletesResult;
			} else {
				indexWriter.abort();
				return null;
			}
		} catch (Throwable throwable) {
			transaction.failed(this, throwable);
		}
		return null;
	}

	public class WriteableIndexWriter extends InstantiatedIndexWriter {
		public WriteableIndexWriter(InstantiatedIndex index) throws IOException {
			super(index);
		}
	}
}
